package aos.voting;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Vector;

import aos.voting.entity.Configuration;
import aos.voting.entity.Data;
import aos.voting.entity.DataSet;
import aos.voting.entity.GlobalVariables;
import aos.voting.entity.Message;
import aos.voting.entity.MessageType;
import aos.voting.exception.AosException;
import aos.voting.log.LogUtils;

public class ServerNode extends NetworkNode {

	protected Vector<Integer> grantTo;
	protected DataSet dataset;
	protected Vector<Queue<Message>> pendingMessage;
	protected boolean isDown;
	protected boolean isRecovering;
	private HashMap<Integer, Integer> clientId2TS;

	public ServerNode(Configuration config) throws AosException {
		super(config);
		grantTo = new Vector<Integer>();
		dataset = new DataSet(GlobalVariables.objectNum);
		pendingMessage = new Vector<Queue<Message>>();
		isDown = false;
		isRecovering = false;
		for (int i = 0; i < GlobalVariables.objectNum; ++i) {
			grantTo.add(-1);
			Queue<Message> q = new LinkedList<Message>();
			pendingMessage.add(q);
		}
		this.clientId2TS = new HashMap<Integer, Integer>();
	}

	protected void increase(Message reqMsg, Message commitMsg) {
		dataset.increase(reqMsg.getObjectId(), reqMsg.getValue(),
				reqMsg.getFromId(), reqMsg.getMsgId());
		LogUtils.keyLog("nodeId: " + nodeId + " " + dataset.toString());
	}

	protected Data get(int dataId) {
		return dataset.get(dataId);
	}

	protected void enqueueMessage(Message msg) {
		pendingMessage.get(msg.getObjectId()).add(msg);
	}

	protected void removeMessage(int dataId, int clientId) {
		for (Message msg : pendingMessage.get(dataId)) {
			if (msg.getFromId() == clientId) {
				pendingMessage.get(dataId).remove(msg);
				return;
			}
		}
	}

	protected Message getMessage(int clientId) {
		for (Queue<Message> queue : pendingMessage) {
			for (Message msg : queue) {
				if (msg.getFromId() == clientId) {
					return msg;
				}
			}
		}
		return null;
	}

	protected Message peekMessage(int dataId) {
		if (pendingMessage.get(dataId).isEmpty()) {
			return null;
		}
		return pendingMessage.get(dataId).peek();
	}

	@Override
	protected void handleMessageReceive(Message msg) throws AosException {
		if (isDown && msg.getMsgType() != MessageType.REACTIVATE) {
			LogUtils.debugLog("rejectMessageReceive: " + msg.toString());
			return;
		} else {
			LogUtils.debugLog("handleMessageReceive: " + msg.toString());
		}

		if (isRecovering) {
			if (msg.getMsgType() == MessageType.RECOVER_RESP) {
				dataset = msg.getDataset();
				isRecovering = false;
			} else if (msg.getFromId() >= config.getNS()) {
				clientId2TS.put(msg.getFromId(), msg.getMsgId());
				if (clientId2TS.size() == config.getNC()) {
					for (int serverId = 0; serverId < config.getNS(); ++serverId) {
						Message recoverMsg = new Message(MessageType.RECOVER,
								serverId, nodeId);
						recoverMsg.setDataset(dataset);
						try {
							Thread.sleep(2000);
						} catch (InterruptedException e) {

						}
						sendMessage(msg);
					}
				}
			}
			return;
		}

		if (dataset.getTS(msg.getFromId()) >= msg.getMsgId()) {
			return;
		}

		switch (msg.getMsgType()) {
		case REQ:
			handleReq(msg);
			break;
		case COMMIT:
			handleCommit(msg);
			break;
		case WITHDRAW:
			handleWithdraw(msg);
			break;
		case DEACTIVATE:
			handleDeactivate(msg);
			break;
		case REACTIVATE:
			handleReactivate(msg);
			break;
		default:
		}
	}

	protected void handleReq(Message msg) throws AosException {
		enqueueMessage(msg);
		if (grantTo.get(msg.getObjectId()) == -1) {
			grantTo.set(msg.getObjectId(), msg.getFromId());
			Data data = dataset.get(msg.getObjectId());
			Message respMsg = new Message(MessageType.GRANT, msg.getFromId(),
					nodeId, data.getDataId(), data.getValue() + msg.getValue(),
					data.getVersion() + 1);
			respMsg.setMsgId(msg.getMsgId());
			sendMessage(respMsg);
		}
	}

	protected void processNext(Message queueMsg) throws AosException {
		if (queueMsg == null) {
			return;
		}
		removeMessage(queueMsg.getObjectId(), queueMsg.getFromId());
		Message ackMsg = new Message(MessageType.ACK, queueMsg.getFromId(),
				nodeId);
		ackMsg.setMsgId(queueMsg.getMsgId());
		sendMessage(ackMsg);
		if (grantTo.get(queueMsg.getObjectId()) == queueMsg.getFromId()) {
			grantTo.set(queueMsg.getObjectId(), -1);
			Message nextMsg = peekMessage(queueMsg.getObjectId());
			if (nextMsg != null) {
				grantTo.set(nextMsg.getObjectId(), nextMsg.getFromId());
				Data data = dataset.get(nextMsg.getObjectId());
				Message respMsg = new Message(MessageType.GRANT,
						nextMsg.getFromId(), nodeId, data.getDataId(),
						data.getValue() + nextMsg.getValue(),
						data.getVersion() + 1);
				respMsg.setMsgId(nextMsg.getMsgId());
				sendMessage(respMsg);
			}
		}
	}

	protected void handleRecover(Message msg) throws AosException {
		if (dataset.compareTo(msg.getDataset()) > 0) {
			Message respMsg = new Message(MessageType.RECOVER_RESP,
					msg.getFromId(), nodeId);
			respMsg.setDataset(dataset);
			sendMessage(respMsg);
		}
	}

	protected void handleCommit(Message msg) throws AosException {
		Message queueMsg = getMessage(msg.getFromId());
		increase(queueMsg, msg);
		processNext(queueMsg);
	}

	protected void handleWithdraw(Message msg) throws AosException {
		Message queueMsg = getMessage(msg.getFromId());
		processNext(queueMsg);
	}

	protected void handleDeactivate(Message msg) throws AosException {
		for (int i = 0; i < grantTo.size(); ++i) {
			grantTo.set(i, -1);
		}
		for (int i = 0; i < pendingMessage.size(); ++i) {
			pendingMessage.get(i).clear();
		}
		isDown = true;
	}

	protected void handleReactivate(Message msg) throws AosException {
		isDown = false;
		isRecovering = true;
		clientId2TS.clear();
	}

	public static void main(String args[]) throws Exception {
		GlobalVariables.initialize("global.txt");
		String logFile = null;
		if (GlobalVariables.logfile != null
				&& !GlobalVariables.logfile.trim().equals("")) {
			logFile = InetAddress.getLocalHost().getHostName() + "_"
					+ GlobalVariables.logfile;
		}
		LogUtils.initialize(GlobalVariables.loglevel, logFile,
				GlobalVariables.logAppend);
		Configuration config = new Configuration(GlobalVariables.input);

		ServerNode node = new ServerNode(config);
		new Thread(node).start();
	}
}
